package gupao.concurrency.kms.disruptor;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import gupao.concurrency.kms.RequestListMock;
import gupao.concurrency.kms.RequestModel;
import gupao.concurrency.utils.ThreadPoolService;
import gupao.concurrency.utils.Utils;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public class DisruptorSample {

    static final int CONSUMER_CONCURRENCY = 4;
    static ThreadPoolExecutor PRODUCER_THREAD_POOL = ThreadPoolService.getOrCreateThreadPoolExecutor("producer",
            1, 1);

//    static WaitStrategy strategy = new BusySpinWaitStrategy();
    static WaitStrategy strategy = new YieldingWaitStrategy();

    public static void main(String[] args) {
        List<RequestModel> mockRequests = RequestListMock.mockRequests();
        Disruptor<OrderEvent> disruptor = createDisruptor(ProducerType.SINGLE, strategy);
        disruptor.handleEventsWith(createConsumers(CONSUMER_CONCURRENCY));
        disruptor.start();
        createAndStartProducer(disruptor, mockRequests);
        Utils.sleepIgnoreExp(60000);
    }

    static void createAndStartProducer(Disruptor<OrderEvent> disruptor,List<RequestModel> mockRequests){
        DefaultOrderEventHandler.total = mockRequests.size();
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
        OrderEventProducer eventProducer = new OrderEventProducer(ringBuffer);
        log.info("[Start]");
        PRODUCER_THREAD_POOL.submit(() -> {
            log.info("{} is start running, list size: {}",Thread.currentThread().getName(), mockRequests.size());
            for(RequestModel req : mockRequests){
                eventProducer.onData(req);
            }
            log.info("{} done.",Thread.currentThread().getName());
            ResultCollector.start().setStrategy(strategy.getClass().getSimpleName());
        });
    }

    static DefaultOrderEventHandler[] createConsumers(int num){
        int x = Math.max(1, num);
        DefaultOrderEventHandler[] results = new DefaultOrderEventHandler[x];
        for(int i = 0; i < x; i++){
            results[i] = new DefaultOrderEventHandler();
            ResultCollector.addConsumer();
        }
        return results;
    }

    static Disruptor<OrderEvent> createDisruptor(ProducerType type, WaitStrategy strategy){
        return new Disruptor<>(
                OrderEvent::new,
                1024 * 1024,
                Executors.defaultThreadFactory(),
                type,
                strategy
        );
    }

}
